package com.cn.remote;

import com.cn.coder.MessageDecoder;
import com.cn.coder.MessageEncoder;
import com.cn.common.Pair;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.exception.MyException;
import com.cn.msg.Message;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class RemoteServer extends AbStractNetty implements INettyRemoteService {

    /**
     * 工作线程数
     */
    private int worker_pool_size = 8;


    private ServerBootstrap bootstrap = null;

    private EventLoopGroup boss = null;

    private EventLoopGroup worker = null;

    private ExecutorService defaultExecutorService;

    private ChannelListener channelListener;

    private String address;

    public void registerChannelListener(ChannelListener channelListener) {
        this.channelListener = channelListener;
    }

    public RemoteServer(int worker_pool_size, String address) {

        super(500);

        this.worker_pool_size = worker_pool_size;
        this.address = address;
        this.defaultExecutorService = Executors.newCachedThreadPool();

        this.worker_pool_size = worker_pool_size > 0 ? worker_pool_size : this.worker_pool_size;

        this.bootstrap = new ServerBootstrap();
        boolean linux = Epoll.isAvailable();

        if (linux) {
            this.boss = new EpollEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("boss线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        } else {
            this.boss = new NioEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("boss线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        }

        //工作线程
        if (linux) {
            this.worker = new EpollEventLoopGroup(this.worker_pool_size, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("worker线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        } else {
            this.worker = new NioEventLoopGroup(this.worker_pool_size, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("worker线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        }


        DefaultEventExecutorGroup defaultEventExecutorGroup = new DefaultEventExecutorGroup(this.worker_pool_size, new ThreadFactory() {
            private AtomicInteger atomicInteger = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("ServerCodecThread_%d", this.atomicInteger.incrementAndGet()));
            }
        });

        linux = false;
        this.bootstrap.group(boss, worker);
        //linux用 EpollServerSocketChannel ；window用NioServerSocketChannel
        if (!Epoll.isAvailable()) {
            bootstrap.channel(NioServerSocketChannel.class);
        } else {
            bootstrap.channel(EpollServerSocketChannel.class);
        }

        this.bootstrap.option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, 65535)// 发送缓冲大小
                .childOption(ChannelOption.SO_RCVBUF, 65535);// 接收缓冲大小
        this.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                //如果是自定义消息，那么用以下代码
                ch.pipeline().addLast(defaultEventExecutorGroup,
//                        new LoggingHandler(LogLevel.ERROR),//日志
                        new MessageEncoder(),//编码器
                        new MessageDecoder(),//解码器
                        new IdleStateHandler(0, 0, 6, TimeUnit.SECONDS),//心跳检测，不管读写，只要超过6秒钟，就主动释放客户端占用的资源
                        new HeartHandler(),
                        new ServerHandler()//业务处理
                );

            }
        });

        //该行代码用来检测netty内存泄漏
//        ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.ADVANCED);

    }

    public void start() {
        if (bootstrap == null) throw new RuntimeException("初始化失败！");
        try {
            //绑定端口，开始同步
            String[] addrs = address.split(":");
            if (addrs.length != 2) {
                throw new MyException("监听地址错误！");
            }
            InetSocketAddress inetSocketAddress = new InetSocketAddress(addrs[0], Integer.valueOf(addrs[1]));
            ChannelFuture channel = bootstrap.bind(inetSocketAddress).sync();
            System.out.println("服务已启动，地址：" + inetSocketAddress.getAddress().getHostAddress() + "，端口：" + inetSocketAddress.getPort());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void shutdown() {
        if (this.boss != null) {
            this.boss.shutdownGracefully();
            this.worker.shutdownGracefully();
            this.cleanProcessor();
            System.out.println("服务已关闭！");
        }
    }

    public void registerProcessor(int code, INettyRequestProcessor processor, ExecutorService executorService) {
        if (null == executorService) {
            executorService = this.defaultExecutorService;
        }
        Pair<INettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorService);
        this.processTables.put(code, pair);
    }

    @Override
    public void unRegisterConnect(ChannelHandlerContext ctx) {
        RemoteUtil.close(ctx);
        if (channelListener != null) {
            channelListener.unregisterEvent(ctx);
        }
    }

    @Override
    public void registerConnect(ChannelHandlerContext ctx) {
        if (channelListener != null) {
            channelListener.registerEvent(ctx);
        }
    }

    class ServerHandler extends SimpleChannelInboundHandler<Message> {

        public ServerHandler() {
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Message message) throws Exception {
            process(message, ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            unRegisterConnect(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            unRegisterConnect(ctx);
        }
    }

    class HeartHandler extends ChannelDuplexHandler {
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                IdleState state = ((IdleStateEvent) evt).state();
                if (state.equals(IdleState.ALL_IDLE)) {
                    //关闭通道
                    ctx.channel().close().addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture channelFuture) throws Exception {
                            System.out.println("与失去联系的客户端通道正确关闭！");
                        }
                    });
                }
            }
            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            unRegisterConnect(ctx);
        }

        @Override
        public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
            unRegisterConnect(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            unRegisterConnect(ctx);
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            super.channelUnregistered(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            super.channelActive(ctx);
            registerConnect(ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            super.channelInactive(ctx);
        }
    }
}
